package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo28Partitions {
  // RDD分区数量的决定因素
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo28Partitions") // 设置任务名
    conf.setMaster("local") // 设置Spark的运行方式
    conf.set("spark.default.parallelism", "20")
    // 创建SparkContext --> Spark程序的入口
    val sc: SparkContext = new SparkContext(conf)

    /**
     * 第一个RDD的分区数：
     * 1、如果大于等2 则由切片的数量决定
     * 2、如果小于2 并且在本地环境 则只有一个分区，如果在集群环境则有两个分区
     */

    val sourceRDD: RDD[String] = sc.textFile("Spark/data/partitions")

    println(s"sourceRDD分区数：${sourceRDD.getNumPartitions}")

    // 使用转换算子
    // 不会产生shuffle --> 分区数默认等于上一个RDD的分区数
    val wordsRDD: RDD[String] = sourceRDD.flatMap(_.split(","))

    println(s"wordsRDD分区数：${wordsRDD.getNumPartitions}")

    // 使用分区类算子--> 会产生shuffle
    // 默认也等于上一个RDD的分区数，如果指定了spark.default.parallelism该参数，则分区数等于参数值，但也可以由自己指定(相当于手动指定Reduce任务的数量)
    // 优先级： 手动指定 >> spark.default.parallelism >> 默认等于上一个RDD的分区数
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)
    println(s"groupRDD分区数：${groupRDD.getNumPartitions}")

    val groupRDD2: RDD[(String, Iterable[String])] = wordsRDD.groupBy((word: String) => {
      word
    }, 5)
    println(s"groupRDD2分区数：${groupRDD2.getNumPartitions}")

    // 手动改变分区数量
    /**
     * 1、使用coalesce
     * 2、使用repartition
     * 如何选择？
     * 1、如果需要减少分区直接使用coalesce 不需要产生shuffle
     * 2、如果需要增加分区可以使用repartition 需要产生shuffle
     * 改变分区数量可以改变最终生成的Task的数量，最终可以对任务进行优化
     */
    val coalesceRDD: RDD[(String, Iterable[String])] = groupRDD2.coalesce(3)
    println(s"coalesceRDD分区数：${coalesceRDD.getNumPartitions}")

    val coalesceRDD2: RDD[(String, Iterable[String])] = groupRDD2.coalesce(10)
    println(s"coalesceRDD2分区数：${coalesceRDD2.getNumPartitions}")

    val repartitionRDD: RDD[(String, Iterable[String])] = coalesceRDD.repartition(6)
    println(s"repartitionRDD分区数：${repartitionRDD.getNumPartitions}")

    val repartitionRDD2: RDD[(String, Iterable[String])] = coalesceRDD.repartition(2)
    println(s"repartitionRDD分区数：${repartitionRDD2.getNumPartitions}")


  }

}
